package raft

/*
 主要的修改如下：

 1. 执行了go fmt raft.go命令，对raft.go这个文件进行格式化；
 `
 2. 在struct Raft结构中增加了一个electionTimeout，并且增加了一个resetTimer()
    函数，每次收到AE消息的时候，就调用resetTimer函数。在resetTimer函数中，重新
    用随机数生成electionTimeout。这样，每次的超时时间都是随机的。以前代码的做法
    是，在Raft对象生成的时候，随机生成一次超时时间，以后这个Raft对象一直用这个
    超时时间，这个跟Raft论文要求不符合。
   
 3. 在Make()函数中调用go rf.sendHeartbeat()创建心跳线程，以后这个心跳线程不再
    退出，除非这个Raft对象被Kill。这样，就不会有多个sendHeartbeat线程同时发送
    心跳消息，只会有一个线程发送这个心跳消息。

 4. 在Raft论文的5.4.1最后一段，描述了如何判断哪个server的logs是最新的。判断方法
    是：a、先判断两个server的最后一个entry的term，谁大，谁的log就新；b、如果最后
    一个entry的term一样，那么看index，谁大谁就新。
      >
      > Raft determines which of two logs is more up-to-date
      > by comparing the index and term of the last entries in the
      > logs. If the logs have last entries with different terms, then
      > the log with the later term is more up-to-date. If the logs
      > end with the same term, then whichever log is longer is
      > more up-to-date.
      > 
    我增加了一个函数isPeersLogUptodate()，当收到RV消息的时候，用来判断对方的
    logs是否比自己的更新。
  
  5. 增加了一些调试信息输出。增加了一个rf.Dbg()函数，这个函数的参数跟DPrintf()
     一样，但是每次会输出Raft对象的更详细信息。增加了一个rf.assert()函数，用来
     检查肯定为“真”的那些条件。如果条件不为真，则assert打印错误信息退，这样可以
     更早发现错误。增加了一个setState()，只在这个函数中更改raft.state，这样就
     可以在这个函数打印出哪个位置更改了Raft状态。在一些关键点打印调试信息，
     例如：发送和接收RV和AE消息的时候，修改状态的时候，apply日志entry的时候，
     等等。
  
  6. 我把 Heartbeat(serverTo int,args *AppendEntriesArgs)函数的args参数去掉了，
     在这个函数内部生成AE消息的参数。不过，这个地方似乎不是一个Bug，是我对Golang
     的理解不够深入，所以错把这个认为是Bug了。
     下面这样定义的struct a是有问题的，如果把struct a定义放到循环内部，就没有
     问题了。
     > type aa struct {
     >         a int
     > }
     > 
     > func main() {
     >         a := &aa{a:0}
     >         for i:=0; i<10; i++ {
     >                 a.a = i
     >                 go func(arg *aa) {
     >                         time.Sleep(time.Duration(10) * time.Microsecond)
     >                         fmt.Printf("arg=%v, *arg=%d\n", arg, *arg)
     >                 }(a)
     >         }
     >         time.Sleep(time.Duration(2) * time.Second)
     > }
     > 

*/

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"bytes"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	"6.5840/labgob"
	"6.5840/labrpc"
)

// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 3D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 3D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

// A Go object implementing a single Raft peer.
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	currentTerm int
	votedFor    int
	log         []logEntry

	commitIndex int
	lastApplied int

	nextIndex  []int
	matchIndex []int

	voteaccount int
	state       int

	lasttime    time.Time
	electionTimeout  int64

	applyCh chan ApplyMsg

	// Your data here (3A, 3B, 3C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

}

type logEntry struct {
	Index   int
	Term    int
	Command interface{}
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	var term int
	var isleader bool
	if rf.state == LEADER {
		isleader = true
	} else {
		isleader = false
	}
	term = rf.currentTerm

	// Your code here (3A).
	return term, isleader
}

func (rf *Raft) setState(state int) {
	pos := getPos()
	rf.Dbg("Pos=%s Old=%s New=%s\n", pos, stName[rf.state],
		stName[state])
	rf.state = state
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
// before you've implemented snapshots, you should pass nil as the
// second argument to persister.Save().
// after you've implemented snapshots, pass the current snapshot
// (or nil if there's not yet a snapshot).
func (rf *Raft) persist() {
	// Your code here (3C).
	// Example:
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.currentTerm)
	e.Encode(rf.votedFor)
	e.Encode(rf.log)
	raftstate := w.Bytes()
	rf.persister.Save(raftstate, nil)
}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (3CL.
	// Example:
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var currentTerm int
	var votedFor int
	//var commitIndex int
	var log []logEntry
	if d.Decode(&currentTerm) != nil ||
		d.Decode(&votedFor) != nil ||
		d.Decode(&log) != nil {
		//   error...

	} else {
		rf.currentTerm = currentTerm
		rf.votedFor = votedFor
		rf.log = log
	}
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (3D).

}

// example RequestVote RPC arguments structure.
// field names must start with capital letters!
type RequestVoteArgs struct {
	LastLogTerm  int
	LastLogIndex int
	Term         int
	CandidateID  int
	// Your data here (3A, 3B).
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
	Term        int
	VoteGranted bool

	// Your data here (3A).
}

//判断是否对方（RV消息的发送者）的日志比自己的更新
func (rf *Raft) isPeersLogUptodate(args *RequestVoteArgs) bool {

	//先用自己的最后一条日志的Term跟对方的对比
	lastEntry := rf.log[len(rf.log)-1]
	if lastEntry.Term > args.LastLogTerm {
		return false
	}

	if lastEntry.Term < args.LastLogTerm {
		return true
	}

	//如果自己最后一条日志的Term与对方相等，则比较谁的Log更多
	if len(rf.log) - 1 > args.LastLogIndex {
		return false
	}

	return true
}


// example RequestVote RPC handler.处理收到的requestvote请求
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {

	rf.mu.Lock()
	defer rf.mu.Unlock()
	rf.Dbg("MSG.RecvRV req: args=%v\n", args)
	reply.Term = rf.currentTerm
	reply.VoteGranted = false
	if args.Term < rf.currentTerm {
		rf.Dbg("MSG.SendRV rpl: args=%v\n\t\treply=%v\n",
			*args, *reply)
		return
	}

	if args.Term > rf.currentTerm {
		rf.votedFor = -1
		rf.currentTerm = args.Term
		rf.setState(FOLLOWER)
		rf.persist()
	}

	//走到这里，说明两者的Term一样。如果自己不是FOLLOWER，那么一定给自己
	//投过票，所以返回false是正确的
	if rf.state != FOLLOWER {
		rf.Dbg("MSG.SendRV rpl: args=%v\n\t\treply=%v\n",
			*args, *reply)
		return;
	}

	//虽然自己是FOLLOWER, 但是，给其他server投过票，不是当前这个RV的发送者
	if rf.votedFor != -1 && rf.votedFor != args.CandidateID {
		rf.Dbg("MSG.SendRV rpl: args=%v\n\t\treply=%v\n",
			*args, *reply)
		return;
	}

	last := len(rf.log) - 1
	rf.assert(last>=0)
	rf.assert(rf.currentTerm == args.Term)
	rf.assert(rf.state == FOLLOWER)

	if rf.isPeersLogUptodate(args) {

		rf.votedFor = args.CandidateID
		rf.persist()

		reply.Term = rf.currentTerm
		reply.VoteGranted = true
		rf.resetTimer()
		rf.Dbg("MSG.SendRV rpl: args=%v\n\t\treply=%v\n",
			*args, *reply)
		return
	}

	rf.Dbg("MSG.SendRV rpl: args=%v\n\t\treply=%v\n", *args, *reply)
	return
	// Your code here (3A, 3B)
}

// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.

func ElapsedUs(t0 time.Time) int64 {
	t1 := time.Now()
	td := t1.Sub(t0)
	us := td.Microseconds()
	return us
}

func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	t0 := time.Now()
	rf.Dbg("RPC.SendRV: to=%d\n\t\targs=%v\n", server, args)
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	if ok {
		td := ElapsedUs(t0)
		rf.Dbg("RPC.RecvRV: %dus, from=%d\n\t\targs=%v, reply=%v\n",
			td, server, args, reply)
	}
	return ok
}

type AppendEntriesArgs struct {
	Term         int
	LeaderID     int
	PrevLogindex int
	PrevLogterm  int
	Entries      []logEntry

	CommittedIndex int
}

type AppendEntriesReply struct {
	Term    int
	Success bool
	XTerm   int
	XIndex  int
	XLen    int
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm
		reply.Success = false
		rf.persist()
		return
	}

	rf.resetTimer()

	if args.Term > rf.currentTerm {
		rf.currentTerm = args.Term
		rf.votedFor = -1
		rf.setState(FOLLOWER)
	}

	isConflict := false
	if args.PrevLogindex >= len(rf.log) {
		reply.XTerm = -1
		reply.XLen = len(rf.log)
		isConflict = true
	} else if args.PrevLogindex >= 0 &&
			args.PrevLogindex <= len(rf.log)-1 &&
			rf.log[args.PrevLogindex].Term != args.PrevLogterm {
		reply.XTerm = rf.log[args.PrevLogindex].Term
		i := args.PrevLogindex
		for rf.log[i].Term == reply.XTerm {
			i -= 1
		}
		reply.XIndex = i + 1
		isConflict = true
	}

	if isConflict {
		reply.Term = rf.currentTerm
		reply.Success = false
		rf.persist()
		return
	}

	for idx, log := range args.Entries {
		ridx := args.PrevLogindex + 1 + idx
		if ridx < len(rf.log) && rf.log[ridx].Term != log.Term {
			// 某位置发生了冲突, 覆盖这个位置开始的所有内容
			rf.log = rf.log[:ridx]
			rf.log = append(rf.log, args.Entries[idx:]...)
			break
		} else if ridx == len(rf.log) {
			// 没有发生冲突但长度更长了, 直接拼接
			rf.log = append(rf.log, args.Entries[idx:]...)
			break
		}
	}

	if args.CommittedIndex > rf.commitIndex {
		if len(rf.log)-1 >= args.CommittedIndex {
			rf.commitIndex = args.CommittedIndex
		} else {
			rf.commitIndex = len(rf.log) - 1
		}
		go rf.CommitChecker()

	}
	reply.Success = true
	reply.Term = rf.currentTerm
	rf.persist()
	return
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs,
			reply *AppendEntriesReply) bool {
	t0 := time.Now()
	rf.Dbg("RPC.SendAE req: to=%d\n\t\targs=%s\n", server, args.Sprint())
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	if ok {
		td := ElapsedUs(t0)
		rf.Dbg("RPC.RecvAE rpl: %dus, from=%d\n\t\targs=%s\n\t\treply=%v\n",
			td, server, args.Sprint(), reply)
	}
	return ok
}

// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if rf.state != LEADER {
		return -1, -1, false
	}
	entry := &logEntry{
		Term:    rf.currentTerm,
		Command: command,
		Index:   len(rf.log),
	}

	rf.log = append(rf.log, *entry)
	//TODO: 这里应该立即触发发送心跳

	return len(rf.log) - 1, rf.currentTerm, true

	// Your code here (3B).
}

// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) resetTimer() {
	rf.electionTimeout = rand.Int63()%150 + 200
	rf.lasttime = time.Now()
}

func (rf *Raft) ticker() {
	for rf.killed() == false {
		rf.mu.Lock()

		tmo := time.Duration(rf.electionTimeout)*time.Millisecond

		if rf.state != LEADER {
			if time.Since(rf.lasttime) > tmo {
				//rf.Dbg("server%d try to start elect", rf.me)
				rf.elect()
			}
		}
		rf.mu.Unlock()

		// Your code here (3A)
		// Check if a leader election should be started.

		// pause for a random amount of time between 50 and 350
		// milliseconds.
		ms := 10
		time.Sleep(time.Duration(ms) * time.Millisecond)
	}
	return
}

func (rf *Raft) elect() {
	rf.setState(CANDIDATE)
	rf.currentTerm += 1
	rf.voteaccount = 1		//首先给自己投一票
	rf.votedFor = rf.me
	rf.resetTimer()
	for i := 0; i < len(rf.peers); i++ {
		if i == rf.me {
			continue
		}
		go rf.collectvote(i)
	}
	return
}

func (rf *Raft) collectvote(serverTo int) {

	rf.mu.Lock()
	defer rf.mu.Unlock()
	args := &RequestVoteArgs{
		Term:         rf.currentTerm,
		CandidateID:  rf.me,
		LastLogIndex: 0,
		LastLogTerm:  0,
	}
	if len(rf.log) > 0 {
		args.LastLogIndex = len(rf.log) - 1
		args.LastLogTerm = rf.log[len(rf.log)-1].Term
	}

	ans := rf.getanswer(serverTo, args)

	if !ans {
		return
	}

	if rf.voteaccount > len(rf.peers)/2 {
		return
	}

	rf.voteaccount += 1
	if rf.voteaccount > len(rf.peers)/2 {
		for i := 0; i < len(rf.peers); i++ {
			rf.nextIndex[i] = len(rf.log)
			rf.matchIndex[i] = 0
		}

		//sendHeartbeat线程一直运行，只要状态改为leader，就开始发送心跳
		rf.setState(LEADER)
		rf.persist()
	}
	return
}

func (rf *Raft) getanswer(serverTo int, args *RequestVoteArgs) bool {
	reply := &RequestVoteReply{}

	rf.mu.Unlock()
	ok := rf.sendRequestVote(serverTo, args, reply)
	rf.mu.Lock()
	if ok == false {
		return false
	}
	if args.Term != rf.currentTerm {
		return false
	}

	if reply.Term > rf.currentTerm {
		rf.currentTerm = reply.Term
		rf.setState(FOLLOWER)
		rf.votedFor = -1
		return false
	}

	return reply.VoteGranted
}

func (rf *Raft) sendHeartbeat() {
	rf.mu.Lock()
	for rf.killed() == false {
		//如果不是leader，那么就空转，什么也不干

		if rf.state != LEADER  {
			rf.mu.Unlock()
			time.Sleep(time.Duration(10) * time.Millisecond)
			rf.mu.Lock()
			continue
		}

		for i := 0; i < len(rf.peers); i++ {
			if i == rf.me {
				continue
			}

			n := len(rf.log)
			if rf.nextIndex[i] > n {
				rf.nextIndex[i] = n
			}

			go rf.Heartbeat(i)
		}
		rf.persist()
		rf.mu.Unlock()
		time.Sleep(time.Duration(50) * time.Millisecond)
		rf.mu.Lock()
	}
	rf.mu.Unlock()
	return
}

func (rf *Raft) Heartbeat(serverTo int) {

	rf.mu.Lock()
	i := serverTo
	reply := &AppendEntriesReply{}
	args := &AppendEntriesArgs{
		Term:         rf.currentTerm,
		LeaderID:     rf.me,
		PrevLogindex: rf.nextIndex[i] - 1,
		CommittedIndex: rf.commitIndex,
	}
	if rf.nextIndex[i]-1 >= 0 {
		args.PrevLogterm = rf.log[rf.nextIndex[i]-1].Term
	}
	if len(rf.log)-1 >= rf.nextIndex[i] {
		args.Entries = rf.log[rf.nextIndex[i]:]
	} else {
		args.Entries = make([]logEntry, 0)
	}
	if rf.state != LEADER {
		rf.mu.Unlock()
		return
	}
	rf.mu.Unlock()

	ok := rf.sendAppendEntries(serverTo, args, reply)

	if !ok {
		return
	}

	rf.mu.Lock()

	if args.Term != rf.currentTerm {
		rf.mu.Unlock()
		return
	}
	if reply.Success == true {
		rf.matchIndex[serverTo] = args.PrevLogindex + len(args.Entries)
		rf.nextIndex[serverTo] = rf.matchIndex[serverTo] + 1

		N := len(rf.log) - 1

		for N > rf.commitIndex && rf.log[N].Term == rf.currentTerm {
			tmp := 1
			for i := 0; i < len(rf.peers); i++ {
				if i == rf.me {
					continue
				}
				if rf.matchIndex[i] >= N && rf.log[N].Term == rf.currentTerm {
					tmp += 1
				}
			}
			if tmp > len(rf.peers)/2 {
				rf.commitIndex = N
				go rf.CommitChecker()
				break
			}
			N -= 1
		}
		rf.mu.Unlock()
		return
	}
	if reply.Term > rf.currentTerm {
		rf.currentTerm = reply.Term
		rf.setState(FOLLOWER)
		rf.votedFor = -1
		rf.resetTimer()
		rf.mu.Unlock()
		return
	}
	if reply.Term == rf.currentTerm && rf.state == LEADER {

		if reply.XTerm == -1 {
			//PrevLogIndex这个位置在Follower中不存在
			rf.nextIndex[serverTo] = reply.XLen
			rf.mu.Unlock()
			return
		}

		i := rf.nextIndex[serverTo] - 1
		for i > 0 && rf.log[i].Term > reply.XTerm {
			i -= 1
		}
		if rf.log[i].Term == reply.XTerm {
			//之前PrevLogIndex发生冲突位置时, Follower的Term自己也有
			rf.nextIndex[serverTo] = i + 1
		} else {
			rf.nextIndex[serverTo] = reply.XIndex
		}
		rf.mu.Unlock()
		return
	}
}

func (rf *Raft) CommitChecker() {
	rf.mu.Lock()
	for rf.commitIndex > rf.lastApplied {
		rf.lastApplied += 1
		msg := &ApplyMsg{
			CommandValid: true,
			Command:      rf.log[rf.lastApplied].Command,
			CommandIndex: rf.lastApplied,
		}
		rf.Dbg("apply: me=%d, idx=%d, %s\n", rf.me, rf.lastApplied,
			SprintEntry(&rf.log[rf.lastApplied]))
		rf.applyCh <- *msg
	}
	rf.mu.Unlock()
	return
}

// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{
		peers:     peers,
		persister: persister,
		me:        me,

		state:       0,
		lasttime:    time.Now(),
		voteaccount: 0,

		votedFor:    -1,
		lastApplied: 0,
		commitIndex: 0,
	}
	rf.applyCh = applyCh

	lo := logEntry{
		Term:  0,
		Index: 0,
	}
	rf.log = make([]logEntry, 0)
	rf.log = append(rf.log, lo)

	if me==0 {
		chLog = make(chan string, 1000)
		go dumpLog()
	}

	// Your initialization code here (3A, 3B, 3C).

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	rf.nextIndex = make([]int, len(rf.peers))
	for i := 0; i < len(rf.nextIndex); i++ {
		rf.nextIndex[i] = len(rf.log)
	}
	rf.matchIndex = make([]int, len(rf.peers))

	rf.resetTimer()

	// start ticker goroutine to start elections
	go rf.ticker()
	go rf.sendHeartbeat()

	return rf
}

